package com.tsing.ipcliveserver.websocket;

import com.tsing.ipcliveserver.util.TranscodingCommandUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * WS-FLV控制器
 *
 * @author TheTsing
 */
public class MessageHandler implements WebSocketHandler {

    @Value("${ws-flv.max-connections:8}")
    private int wsFlvMaxConnections;

    @Autowired
    private AsyncTranscodeHandler asyncTranscodeHandler;

    private final Map<String, Process> processMap = new ConcurrentHashMap<>();

    private static final AtomicInteger CONNECTION_COUNT = new AtomicInteger(0);

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        if (CONNECTION_COUNT.incrementAndGet() > wsFlvMaxConnections) {
            try {
                session.close(CloseStatus.POLICY_VIOLATION);
            } catch (Exception ignored) {
            }
            return;
        }
        transcode(session);
    }

    @Override
    public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) {
    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) {
        try {
            session.close(CloseStatus.SERVER_ERROR);
        } catch (Exception ignored) {
        }
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        CONNECTION_COUNT.decrementAndGet();
        Process process = processMap.remove(session.getId());
        if (process != null) {
            process.destroy();
        }
    }

    @Override
    public boolean supportsPartialMessages() {
        return false;
    }

    private void transcode(WebSocketSession session) {
        Process process;
        try {
            String cmd = TranscodingCommandUtil.generateTranscodingCommand(session.getUri().getRawQuery());
            process = Runtime.getRuntime().exec(cmd);
        } catch (Exception e) {
            try {
                session.close(CloseStatus.POLICY_VIOLATION);
            } catch (Exception ignored) {
            }
            return;
        }
        processMap.put(session.getId(), process);
        asyncTranscodeHandler.handlerInputStream(session, process);
        asyncTranscodeHandler.handlerErrorStream(process);
    }

}
